package com.hrt.test.common;

import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import com.hrt.datasender.mq.model.MqModel;

public class MsgQueue {

    public String guid;

    /**
     * 队列对象
     */
    private BlockingQueue<Object> queue;
    /**
     * 队列最大大小
     */
    private final Integer maxQueueSize = 90000;
    /**
     * 消息监听器
     */
    private IMsgListener listener;
    /**
     * 队列名称
     */
    private String queueName;

    public MsgQueue(String queueName) {
        System.out.println("初始化队列：" + queueName);
        guid = UUID.randomUUID().toString();
        this.queueName = queueName;
        queue = new LinkedBlockingQueue<Object>(maxQueueSize);
    }

    /**
     * 消息入队
     * 
     * @param msg
     * @throws InterruptedException
     */
    public void produce(Object msg) {
        try {
            queue.put(msg);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public synchronized void consume(IMsgListener listener)
            throws QueueHasBeenListenedException, NullQueueListenerException {
        if (this.listener != null) {
            throw new QueueHasBeenListenedException(queueName);
        }

        if (listener == null) {
            throw new NullQueueListenerException(queueName);
        }

        this.listener = listener;
        startListener();
    }

    private void startListener() {
        if (listener == null) {
            return;
        }

        new Thread() {

            public void run() {
                while (true) {
                    try {
                        Object msg = queue.take();
                        listener.onReceive(msg);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }.start();
    }

    public synchronized MqModel consume() throws QueueHasBeenListenedException, NullQueueListenerException {
        MqModel msg = null;
        // 判断是否有监听器监听该队列，若有则禁止后续读取
        if (this.listener != null) {
            return msg;
        }

        try {
            msg = (MqModel) queue.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return msg;
    }

    // 获取队列大小
    public int size() {
        return queue.size();
    }
}
